package com.atguigu.actual.edu0417.app.dwd.db;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.beans.BaseDbTableProcess;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.func.BaseDbTableProcessFunction;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;

/**
 * @author: 洛尘
 * @since: 2023-10-20 15:47
 * @description: 动态分流基本事实表
 **/
public class BaseDbApp extends BaseApp {
    public static void main(String[] args) {
        BaseDbApp baseDbApp = new BaseDbApp();
        baseDbApp.base(args);
    }

    @Override
    public void process(StreamExecutionEnvironment env) {
        String groupID="dwd_base_db_group";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(EduConfig.KAFKA_TOPIC_DB, groupID);

        SingleOutputStreamOperator<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source_basedb");
//        kafkaDS.print(">>>>");
        //清洗数据 类型转换jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                            String type = jsonObj.getString("type");
                            if (!type.startsWith("bootstrap-")) {
                                //过滤掉维度数据
                                out.collect(jsonObj);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
        );
//        jsonDS.print(">>>");
        //从配置表中读取数据
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop101")
                .port(3306)
                .databaseList("edu_config")
                .tableList("edu_config.table_process_dwd")
                .username("root")
                .password("000000")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
        SingleOutputStreamOperator<String> mysqlDwdDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc_mysql_dwd_source");
/*
##:4> {"before":null,"after":{"source_table":"comment_info","source_type":"insert","sink_table":"dwd_interaction_comment_add","sink_columns":"id,user_id,chapter_id,course_id,comment_txt,deleted"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process_dwd","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1697817097899,"transaction":null}
###:4> {"before":null,"after":{"source_table":"favor_info","source_type":"insert","sink_table":"dwd_interaction_favor_add","sink_columns":"id,course_id,user_id,deleted"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process_dwd","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1697817097900,"transaction":null}
###:4> {"before":null,"after":{"source_table":"review_info","source_type":"insert","sink_table":"dwd_interaction_review_add","sink_columns":"id,user_id,course_id,review_txt,review_stars,deleted"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process_dwd","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1697817097900,"transaction":null}
###:4> {"before":null,"after":{"source_table":"user_info","source_type":"insert","sink_table":"dwd_user_register","sink_columns":"id,create_time"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"edu_config","sequence":null,"table":"table_process_dwd","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1697817097900,"transaction":null}
 */
        //将读取的数据读取到广播流中
        MapStateDescriptor<String, BaseDbTableProcess> broadCastDes = new MapStateDescriptor<>("broadCastDes", String.class, BaseDbTableProcess.class);
        BroadcastStream<String> broadcastDS = mysqlDwdDS.broadcast(broadCastDes);

        //将主流和广播流进行关联
        BroadcastConnectedStream<JSONObject, String> connectDS = jsonDS.connect(broadcastDS);

        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(
                new BaseDbTableProcessFunction(broadCastDes)
        );
//        realDS.print(">>>");
        //将动态分流数据写入不同的主题中
        realDS.sinkTo(
                MyKafkaUtil.getKafkaSinkBySchema(
                        new KafkaRecordSerializationSchema<JSONObject>() {
                            @Nullable
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObject, KafkaSinkContext kafkaSinkContext, Long aLong) {
                                String sinkTable = jsonObject.getString("sink_table");
                                //在这里删除的原因是因为在处理主流数据的时候为了补充数据获取主题，但是传递数据不能传递下去，所以需要删除
                                jsonObject.remove("sink_table");
                                return new ProducerRecord<byte[], byte[]>(sinkTable,jsonObject.toString().getBytes());
                            }
                        }
                )
        );

    }
}